Populate dag and note before firing dag run state-change listeners#69873
Populate dag and note before firing dag run state-change listeners#69873kacpermuda wants to merge 2 commits into
Conversation
kacpermuda
requested review from
bugraoz93,
choo121600,
ephraimbuddy,
henry3260,
jason810496,
pierrejeambrun,
rawwar and
shubhamraj-git
as code owners
| # Sort keys so that the "note" update happens before "state" update, so that the listeners called inside | ||
| # `patch_dag_run_state()` already receive updated DagRun object with note. | ||
| for attr_name, attr_value_raw in sorted(data.items()): |
There was a problem hiding this comment.
Using sorted(data.items()) to force "note" before "state" couples correctness to the field names' alphabetical order rather than expressing the ordering intent explicitly. This works today because "note" < "state", but is a trap for the next field added to DAGRunPatchBody & nothing would signal that ordering matters.
I suggest using an explicit ordering or just processing "note" before the loop which would be clearer and safer.
There was a problem hiding this comment.
+1 for this, we might have a similar problem on the Task Instance endpoint as well. (Just realized there is another PR opened for this)
There was a problem hiding this comment.
Thanks, that makes sense, improved this piece of code by removing the loop entirely and just ordering calls explicitly. @pierrejeambrun do you mind linking the taskinstance PR? Would love to track it and make sure these fixes are all merged, included in next release and backported.
pierrejeambrun
left a comment
pierrejeambrun
left a comment
There was a problem hiding this comment.
Thanks overall looking good beside Amogh's comment.
| # Sort keys so that the "note" update happens before "state" update, so that the listeners called inside | ||
| # `patch_dag_run_state()` already receive updated DagRun object with note. | ||
| for attr_name, attr_value_raw in sorted(data.items()): |
There was a problem hiding this comment.
+1 for this, we might have a similar problem on the Task Instance endpoint as well. (Just realized there is another PR opened for this)
kacpermuda
force-pushed
the
dag-listeners-extend
branch
from
5704cd3 to
1cdf0dc
Compare
3 participants
When patching a Dag Run's state to success or failed via the API, two issues
existed with the listener call:
dag_run.dagwas alwaysNonewhen the listener received the object, because the route loads the dag separately but never attached it to the run. After the change, the listener will have access to SerializedDag that is used for example by OpenLineage listener to create an AirflowDagRunFacet.noteandstatewere patched in the same request (dagrun state was set and note was provided), the note could be applied after the listener fired, so the listener saw the old / no note. Fixed by sorting the patch fields sonoteis always processed beforestate.Was generative AI tooling used to co-author this PR?
Generated-by: Claude Sonnet 4.6 following the guidelines
{pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.